package dyyx.batch;

import java.time.LocalDateTime;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;

public class WordCount {

	public static void main(String[] args) throws Exception {
		ParameterTool parameterTool = ParameterTool.fromArgs(args);

		String input = parameterTool.get("input");
		String output = parameterTool.get("output");

		if (input == null) {
			input = "/tmp/flinkdemo/batch/input.txt";
		}
		if (output == null) {
			output = "/tmp/flinkdemo/batch/output";
		}

		System.out.println("input=" + input);
		System.out.println("output=" + output);

		// ExecutionEnvironment vs StreamExecutionEnvironment
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

		DataSet<String> text = env.readTextFile(input);

		DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).groupBy(0).sum(1);

		counts.writeAsCsv(output, "\n", " ");
		
		String job = "flink-demo-batch-wordcount";
		// env.execute(job);
		
		System.out.println(job+" run done,"+LocalDateTime.now());

	}

}
